from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F

from cn.itcast.tag.base.BaseModel import BaseModel
from cn.itcast.tag.bean.ESMeta import ruleToESMeta

"""
-------------------------------------------------
   Description :	TODO：
   SourceFile  :	MarriageModel
   Author      :	itcast team
-------------------------------------------------
"""

# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'

# todo 婚姻状态模型
class MarriageModel(BaseModel):
    def compute(self, es_df, five_df):
        """
        标签计算逻辑实现：基于婚姻状态匹配五级标签
        
        参数:
            es_df: 业务数据DataFrame，包含用户婚姻状态字段
            five_df: 五级标签规则DataFrame，包含id和rule字段
            
        返回:
            DataFrame: 包含userId和tagsId的标签计算结果
        """
        # 调试输出数据结构信息
        es_df.printSchema()
        es_df.show()
        five_df.printSchema()
        five_df.show()

        # 基于婚姻状态字段与标签规则进行左连接
        new_df = es_df.join(
            other=five_df,
            on=es_df['marriage'] == five_df['rule'],  # 按婚姻状态匹配规则
            how='left'  # 使用左连接保留所有用户数据
        ).select(
            es_df['id'].alias("userId"),             # 用户ID
            five_df['id'].alias("tagsId")            # 匹配到的标签ID
        )
        
        return new_df


if __name__ == '__main__':
    ageModel = MarriageModel(66)
    ageModel.execute()